package edu.maxflow.mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Reducer;

import edu.maxflow.Edge;
import edu.maxflow.Path;
import edu.maxflow.Vertex;
import edu.maxflow.mapreduce.util.Accumulator;
import edu.maxflow.mapreduce.util.CounterEnum;
import edu.maxflow.mapreduce.util.MRConstants;

public class MaxFlowReducer extends Reducer<LongWritable, Vertex, LongWritable, Vertex> {
	private long k = 100; // no more than this number of excess path. Increases
							// performance.

	// local accumulators.
	Accumulator augmentedPathAccumulator = new Accumulator();
	Accumulator sourceExcessPathAccumulator = new Accumulator();
	Accumulator targetExcessPathAccumulator = new Accumulator();

	public void reduce(LongWritable key, Iterable<Vertex> vertexIterable, Context context) throws IOException, InterruptedException {
		// get configuration variables
		Configuration conf = context.getConfiguration();
		long targetVertexId = conf.getInt(MRConstants.TARGET_VERTEX_ID, MRConstants.TARGET_VERTEX_DEFAULT_VALUE);
		int round = conf.getInt(MRConstants.ROUND, MRConstants.ROUND_DEFAULT_VALUE);

		Vertex masterVertex = new Vertex();
		List<Path> sourceExcessPaths = new ArrayList<Path>();
		List<Path> targetExcessPaths = new ArrayList<Path>();

		Iterator<Vertex> vertexIterator = vertexIterable.iterator();
		while (vertexIterator.hasNext()) {
			Vertex vertex = vertexIterator.next();

			if (vertex.getAdjacentEdges().size() != 0) {
				masterVertex.getAdjacentEdges().addAll(vertex.getAdjacentEdges());
				masterVertex.getSourceExcessPaths().addAll(vertex.getSourceExcessPaths());
				masterVertex.getTargetExcessPaths().addAll(vertex.getTargetExcessPaths());
				masterVertex.setVertexId(vertex.getVertexId());
			}

			for (Path sourceExcessPath : vertex.getSourceExcessPaths()) {
				if (key.get() == targetVertexId) {
					long flow = augmentedPathAccumulator.acceptAndReturnFlow(sourceExcessPath);

					context.getCounter(CounterEnum.MAXFLOW).increment(flow);
				} else if (sourceExcessPaths.size() < k) {
					if (sourceExcessPathAccumulator.accept(sourceExcessPath)) {
						sourceExcessPaths.add(sourceExcessPath);
					}
				}
			}

			for (Path targetExcessPath : vertex.getTargetExcessPaths()) {
				if (targetExcessPaths.size() < k && targetExcessPathAccumulator.accept(targetExcessPath)) {
					targetExcessPaths.add(targetExcessPath);
				}
			}
		}

		if (masterVertex != null) {
			// TODO :something is wrong
		}

		if (masterVertex.getSourceExcessPaths().size() == 0 && sourceExcessPaths.size() > 0) {
			context.getCounter(CounterEnum.SOURCE_MOVE).increment(1L);
		}

		if (masterVertex.getTargetExcessPaths().size() == 0 && targetExcessPaths.size() > 0) {
			context.getCounter(CounterEnum.SINK_MOVE).increment(1L);
		}

		if (key.get() == targetVertexId) {
			FSDataOutputStream outputStream = null;

			try {
				// write accepted edges to HDFS file
				FileSystem fileSystem = FileSystem.get(conf);

				org.apache.hadoop.fs.Path path = new org.apache.hadoop.fs.Path(MRConstants.PATH_AUGMENTED_EDGE + round);
				outputStream = fileSystem.create(path);

				outputStream.writeInt(augmentedPathAccumulator.getAcceptedEdges().size());
				for (Edge edge : augmentedPathAccumulator.getAcceptedEdges()) {
					edge.write(outputStream);
				}
			} catch (Exception e) {
				System.err.println("File write exception occured");
			} finally {
				if (outputStream != null) {
					outputStream.close();
				}
			}
		}

		// emit vertex as master
		Vertex vertexToEmit = new Vertex();
		vertexToEmit.setVertexId(key.get());
		vertexToEmit.setSourceExcessPaths(sourceExcessPaths);
		vertexToEmit.setTargetExcessPaths(targetExcessPaths);
		vertexToEmit.setAdjacentEdges(masterVertex.getAdjacentEdges());

		context.write(key, vertexToEmit);
	}
}
